package com.cloud.school.app.service.impl;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.serializer.ValueFilter;
import com.cloud.school.app.config.WebSocketConfiguration;
import com.cloud.school.app.model.GroupMemberModel;
import com.cloud.school.app.model.MessageModel;
import com.cloud.school.app.model.UserModel;
import com.cloud.school.app.service.GroupMemberService;
import com.cloud.school.app.service.MessageService;
import com.cloud.school.app.service.UserService;
import com.cloud.school.app.util.enumeration.MessageReceiveType;
import com.cloud.school.app.util.enumeration.MessageStatus;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import org.springframework.web.socket.CloseStatus;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;
import org.springframework.web.socket.handler.TextWebSocketHandler;

import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;

import static com.cloud.school.app.controller.ChatRoomController.CHAT_ROOM_USER_ID_KEY;

/**
 * @ClassName ChatWebSocketHandler
 * @Description TODO
 * @Author 36297
 * @Date 2018/12/11 20:52
 * @Version 1.0
 */
@Slf4j
@Service("chatWebSocketHandler")
public class ChatWebSocketHandler extends TextWebSocketHandler {

    public static final String SESSION_ID_USER_ID_MAP_KEY = "sessionIdUserIdMap";
    public static final String USER_ID_SESSION_ID_MAP_KEY = "userIdSessionIdMap";
    private final static Map<String, WebSocketSession> sessionIdSessionMap = new ConcurrentHashMap();


    @Autowired
    private MessageService messageService;
    @Autowired
    private GroupMemberService groupMemberService;
    @Autowired
    private CacheService cacheService;
    @Autowired
    private UserService userService;

    @Override
    public void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception {
        String msg = message.getPayload();
        if (msg == null) return;
        saveAndSendMessage(msg);
    }

    @Override
    public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception {
        String userId = cacheService.hashGet(SESSION_ID_USER_ID_MAP_KEY, session.getId());
        cacheService.hashDelete(SESSION_ID_USER_ID_MAP_KEY, session.getId());
        cacheService.hashDelete(USER_ID_SESSION_ID_MAP_KEY, userId);
        sessionIdSessionMap.remove(session.getId());
        log.debug("退出聊天室: userOID={}, sessionID={}", userId, session.getId());
    }

    @Override
    public void afterConnectionEstablished(WebSocketSession session) throws Exception {
        Object attr = session.getAttributes().get(WebSocketConfiguration.USER_KEY);
        String userId = attr == null ? null : attr.toString();
        cacheService.hashPut(SESSION_ID_USER_ID_MAP_KEY, session.getId(), userId);
        cacheService.hashPut(USER_ID_SESSION_ID_MAP_KEY, userId, session.getId());
        sessionIdSessionMap.put(session.getId(), session);
        log.debug("进入聊天室: userOID={}, sessionID={}", userId, session.getId());
    }

    @Override
    public void handleTransportError(WebSocketSession session, Throwable exception) throws Exception {
        String userId = cacheService.hashGet(SESSION_ID_USER_ID_MAP_KEY, session.getId());
        log.error("消息接收异常: userOID=" + userId + ", sessionID=" + session.getId(), exception);
    }

    @Async
    public void saveAndSendMessage(String msg) {
        try {
            MessageModel messageModel = JSON.parseObject(msg, MessageModel.class);
            messageModel.setSendTime(System.currentTimeMillis());
            messageModel.setStatus(MessageStatus.DELIVERED.getCode());
            UserModel userModel = userService.queryById(messageModel.getSendUid());
            messageModel.setSender(userModel);
            msg = JSON.toJSONString(messageModel, new ValueFilter() {
                @Override
                public Object process(Object object, String name, Object value) {
                    if (value instanceof Long) {
                        return value.toString();
                    }
                    return value;
                }
            });
            if (MessageReceiveType.codeOf(messageModel.getReceiveType()) == MessageReceiveType.USER) {
                if (!sendUserMessage(messageModel.getReceiveId(), msg)) {
                    messageModel.setStatus(MessageStatus.UNDELIVERED.getCode());
                }
            } else if (MessageReceiveType.codeOf(messageModel.getReceiveType()) == MessageReceiveType.GROUP) {
                if (!sendGroupMessage(messageModel.getReceiveId(), msg)) {
                    messageModel.setStatus(MessageStatus.UNDELIVERED.getCode());
                }
            } else if (MessageReceiveType.codeOf(messageModel.getReceiveType()) == MessageReceiveType.CHAT_ROOM) {
                if (!sendChatRoomMessage(messageModel.getReceiveId(), msg)) {
                    messageModel.setStatus(MessageStatus.UNDELIVERED.getCode());
                }
            }
            messageService.save(messageModel);
        } catch (Exception e) {
            log.error("消息处理异常,消息:{}", msg, e);
        }
    }

    /**
     * 单发
     */
    private void sendMessage(WebSocketSession session, TextMessage message) throws IOException {
        session.sendMessage(message);
    }

    /**
     * 单发
     */
    public boolean sendUserMessage(Long receiveUid, String msg) throws IOException {
        if (receiveUid != null) {
            String sessionId = cacheService.hashGet(USER_ID_SESSION_ID_MAP_KEY, receiveUid.toString());
            if (sessionId != null) {
                WebSocketSession session = sessionIdSessionMap.get(sessionId);
                if (session != null) {
                    TextMessage message = new TextMessage(msg);
                    sendMessage(session, message);
                } else {
                    cacheService.hashDelete(USER_ID_SESSION_ID_MAP_KEY, receiveUid.toString());
                    cacheService.hashDelete(SESSION_ID_USER_ID_MAP_KEY, sessionId);
                    log.info("消息未送达,原因接收人session为null,消息:{}", msg);
                    return false;
                }
            } else {
                log.info("消息未送达,原因接收人不在线,消息:{}", msg);
                return false;
            }
        } else {
            log.info("消息未送达,原因接收人id为空,消息:{}", msg);
            return false;
        }
        return true;
    }

    /**
     * 群发
     */
    public boolean sendGroupMessage(Long receiveGid, String msg) throws IOException {
        if (receiveGid != null) {
            List<Long> userIds = groupMemberService.queryByGroupId(receiveGid).stream().filter(g -> g.getUserId() != null).map(GroupMemberModel::getUserId).collect(Collectors.toList());
            List<String> sIds = userIds.stream().map(uid -> cacheService.hashGet(USER_ID_SESSION_ID_MAP_KEY, uid.toString())).filter(sid -> sid != null).collect(Collectors.toList());
            List<WebSocketSession> sessions = sIds.stream().map(id -> sessionIdSessionMap.get(id)).filter(s -> s != null).collect(Collectors.toList());
            removeSession(sIds);
            TextMessage message = new TextMessage(msg);
            sendMessage(sessions, message);
        } else {
            log.info("消息未送达,原因接收群组id为空,消息:{}", msg);
            return false;
        }
        return true;
    }

    /**
     * 群发
     */
    public boolean sendChatRoomMessage(Long receiveChatRoomid, String msg) throws IOException {
        if (receiveChatRoomid != null) {
            List<Long> userIds = cacheService.setMembers(CHAT_ROOM_USER_ID_KEY + receiveChatRoomid).stream().map(Long::valueOf).collect(Collectors.toList());
            List<String> sIds = userIds.stream().map(uid -> cacheService.hashGet(USER_ID_SESSION_ID_MAP_KEY, uid.toString())).filter(sid -> sid != null).collect(Collectors.toList());
            List<WebSocketSession> sessions = sIds.stream().map(id -> sessionIdSessionMap.get(id)).filter(s -> s != null).collect(Collectors.toList());
            removeSession(sIds);
            TextMessage message = new TextMessage(msg);
            sendMessage(sessions, message);
            log.debug("消息已推送,接收聊天室id:{},消息:{}", receiveChatRoomid, msg);
        } else {
            log.info("消息未送达,原因接收聊天室id为空,消息:{}", msg);
            return false;
        }
        return true;
    }

    /**
     * 群发
     */
    private void sendMessage(List<WebSocketSession> sessions, TextMessage message) throws IOException {
        for (WebSocketSession session : sessions) {
            session.sendMessage(message);
        }
    }

    private void removeSession(List<String> sIds) {
        sIds.forEach(sId -> {
            WebSocketSession session = sessionIdSessionMap.get(sId);
            if (session == null) {
                String userId = cacheService.hashGet(SESSION_ID_USER_ID_MAP_KEY, sId);
                cacheService.hashDelete(SESSION_ID_USER_ID_MAP_KEY, sId);
                cacheService.hashDelete(USER_ID_SESSION_ID_MAP_KEY, userId);
            }
        });
    }
}
